package org.example.flinktest.apidemo;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/12 16:06
 */
public class ProcessFunctionDemo {

  public static void main(String[] args) throws Exception {
    //环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    //设置时间触发
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //source
    DataStream<String> source =env.socketTextStream("10.0.75.2",7777);

    //transform
    DataStream<SensorReading> dataStream=source
        .map(new MakeSensorMapFunction())
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
              @Override
              public long extractTimestamp(SensorReading element) {
                return element.getTimestamp()*1000;
              }
            });

    //如何10秒内温度不断的上升的将告警
    DataStream<String> alarmStream = dataStream
        .keyBy("id")
        .process(new TemperatureRiseAlarm());


    //sink
    alarmStream.print("alarm data");
    dataStream.print("input data");


    env.execute("window job test");
  }

}


class  TemperatureRiseAlarm extends KeyedProcessFunction<Tuple, SensorReading, String> {
  private ValueState<Double> latestTemp;
  private ValueState<Long> currentTimer;

  @Override
  public void open(Configuration parameters) throws Exception {
    // 定义一个状态，用来保存上一个数据的温度值
    latestTemp=getRuntimeContext().getState(new ValueStateDescriptor<Double>("latestTemp",Double.class));
    currentTimer=getRuntimeContext().getState(new ValueStateDescriptor<Long>("currentTimer",Long.class));
  }


  @Override
  public void processElement(SensorReading value, Context ctx, Collector<String> out)throws Exception{
    //之前的温度
    Double preTemp =latestTemp.value();
    //跟新温度
    latestTemp.update(value.getTemperature());
    //获取定时器的时间
    Long curTimerTs= currentTimer.value();
    System.out.println(ctx.timerService().currentWatermark());
    if(preTemp==null){

    }else if (value.getTemperature()> preTemp && curTimerTs==null) {
      //温度上升且没有设过定时器，注册定时器，并更新状态
      long timeTs = ctx.timerService().currentWatermark()+10000L;
      ctx.timerService().registerEventTimeTimer(timeTs);
      currentTimer.update(timeTs);
    } else if(value.getTemperature()<preTemp && curTimerTs!=null) {
      //10秒内温度有回落就删除定时器,并清空状态
      ctx.timerService().deleteEventTimeTimer(curTimerTs);
      currentTimer.clear();
    }
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
    //直接输出告警
    out.collect(ctx.getCurrentKey()+"温度持续上升");
    currentTimer.clear();
  }
}
